package com.atguigu.app.dwd.db;

import com.atguigu.util.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

//TODO 对订单表进行处理过滤，得到所需要的表
public class DwdCourseOrderDetail {
    public static void main(String[] args) {
          StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
                  env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // TODO 2 设置状态后端
                  /*
                  env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
                  env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
                  env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
                  env.setStateBackend(new HashMapStateBackend());
                  env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
                  System.setProperty("HADOOP_USER_NAME", "atguigu");
                   */
        String topicId = "topic_db";
            String groupId = "dwd_course_order_detail";
        //TODO 得到topic_db数据
        tableEnv.executeSql("create table topic_db(\n" +
                "`table` String,\n" +
                "`type` String,\n" +
                "`ts` BIGINT,\n" +
                "`data` Map<String,String>,\n" +
                "`old` Map<String,String>,\n" +
                " `pt` as proctime() )"
                +KafkaUtil.getKafkaDDL(topicId,groupId));

        //TODO 得到order_info数据
        Table orderInfo = tableEnv.sqlQuery("select \n" +
                    "data['id'] id,\n" +
                    "data['user_id'] user_id,\n" +
                    "data['final_amount'] final_amount,\n" +
                    "data['order_status'] order_status,\n" +
                    "data['province_id'] province_id,\n" +
                    "data['session_id'] sid,\n" +
                    "`ts`,\n" +
                    "`pt` \n" +
                    " from topic_db \n" +
                " where `table` = 'order_info' and `type` = 'insert' " );
        tableEnv.createTemporaryView("order_info",orderInfo);
        //TODO 得到order_detail数据
        Table orderDetail = tableEnv.sqlQuery("select \n" +
                "`data`['id'] id,\n" +
                "`data`['course_id'] course_id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['final_amount'] final_amount,\n" +
                "ts,\n" +
                "pt\n" +
                "from topic_db\n" +
                "where `table` = 'order_detail' and `type` = 'insert'");
            tableEnv.createTemporaryView("order_detail",orderDetail);
  //TODO JOIN表格
        Table orderDetailFullTable = tableEnv.sqlQuery("select \n" +
                "od.id,\n" +
                "od.course_id,\n" +
                "od.order_id,\n" +
                "od.user_id,\n" +
                "od.final_amount,\n" +
                "oi.order_status,\n" +
                "oi.province_id,\n" +
                "oi.sid,\n" +
                "od.ts,\n" +
                "od.pt,\n" +
                "UNIX_TIMESTAMP() rt\n" +
                "from order_detail od\n" +
                "join order_info oi\n" +
                "on od.order_id = oi.id" );
        tableEnv.createTemporaryView("order_detail_full",orderDetailFullTable);
        tableEnv.executeSql("create table dwd_course_order_detail (\n" +
                "id STRING,\n" +
                "course_id STRING,\n" +
                "order_id STRING,\n" +
                "user_id STRING,\n" +
                "final_amount STRING,\n" +
                "order_status STRING,\n" +
                "province_id STRING,\n" +
                "sid STRING,\n" +
                "ts BIGINT,\n" +
                "pt TIMESTAMP_LTZ(3),\n" +
                "rt BIGINT, \n" +
                "PRIMARY KEY(id) NOT ENFORCED \n" +
                ")" + KafkaUtil.getUpsertKafkaSinkDDL(groupId));
//        tableEnv.sqlQuery("select * from order_detail_full").execute().print();
        tableEnv.executeSql("insert into dwd_course_order_detail select * from order_detail_full");
    }
}
